package com.zhiyou.bd14;
//把文件加下的小文件合并成大文件

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

import org.apache.avro.Schema;
import org.apache.avro.file.DataFileWriter;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.specific.SpecificDatumWriter;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.fs.FileUtil;

import com.zhiyou.bd14.schema.SmallFile;

public class AvroMergeSmallFile {
	private Schema.Parser parser = new Schema.Parser();
	private Schema schema;
	//用来设置添加要合并文件的名称
	private List<String> inputFilePaths =new ArrayList<String>();
	//在构造方法中初始化schame
	public AvroMergeSmallFile(){
		schema = SmallFile.getClassSchema();
	}
	//添加要合并的文件夹
	public void addInputFileDir(String inputDir) throws IOException{
		//获取当前文件夹下的缩有文件
		File[] files =FileUtil.listFiles(new File(inputDir));
		//获取当前文件夹下的所有文件路径
		for (File file : files) {
			inputFilePaths.add(file.getPath());
		}
	}
	//把inputFilePaths中的所有文件合并到一个avro文件中
	public void mergeFile(String outPutPath) throws Exception{
		DatumWriter<SmallFile> write = new SpecificDatumWriter<SmallFile>();
		DataFileWriter<SmallFile> fileWriter = new DataFileWriter<SmallFile>(write);
		fileWriter.create(SmallFile.getClassSchema(), new File(outPutPath));
		//intputFilePath的文件一个个读取出来根据模式放入avro文件中[
		for (String filePath : inputFilePaths) {
			File inputFile = new File(filePath);
			//把文件读取成字节数组,放入context
			byte[] context = FileUtils.readFileToByteArray(inputFile);
			//ByteBuffer调用wrap方法把字节数组组装成一个ByteBuffer对象,作为参数设置到oneSmallFile的content的属性中
			SmallFile oneSmallFile = SmallFile.newBuilder().setFileName(inputFile.getAbsolutePath()).
					setContent(ByteBuffer.wrap(context)).build();
			fileWriter.append(oneSmallFile);
			System.out.println("写入"+inputFile.getAbsolutePath()+"成功");
			}
		fileWriter.flush();
		fileWriter.close();
	}
	public static void main(String[] args) throws Exception {
		AvroMergeSmallFile smallFile = new AvroMergeSmallFile();
		smallFile.addInputFileDir("C:\\Users\\ad\\Desktop\\reversetext");
		smallFile.mergeFile("C:\\Users\\ad\\Desktop\\rever");
		
	}
}
